"""
Case Type   : ustore_DQL_SET_OPERATIONS
Case Name   : union与join结合
Create At   : 2022-05-12
Owner       : nanyang12
Description :
    1.创建表
    2.使用copy命令插入数据
    3.union与join结合
    4.环境清理
Expect      :
    1.创建成功
    2.数据插入成功
    3.成功
    4.清理成功
History     :
"""

import os
import subprocess
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Ustore01(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----- {os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_root = Node('PrimaryRoot')
        self.com = Common()
        self.constant = Constant()
        self.table_name1 = 't_ustore_winagg_0001_01'
        self.table_name2 = 't_ustore_winagg_0001_02'
        self.table_name3 = 't_ustore_winagg_0001_03'
        self.table_name4 = 't_ustore_winagg_0001_04'
        self.table_name5 = 't_ustore_winagg_0001_05'
        self.copy_file1 = 'customer_33.txt'
        self.copy_file2 = 'call_center_33.txt'
        self.copy_file3 = 'web_page_33.txt'
        self.copy_file4 = 'promotion_33.txt'
        self.copy_file5 = 'catalog_page_33.txt'
        self.cur_path_1 = os.path.join(macro.SCRIPTS_PATH, self.copy_file1)
        self.cur_path_2 = os.path.join(macro.SCRIPTS_PATH, self.copy_file2)
        self.cur_path_3 = os.path.join(macro.SCRIPTS_PATH, self.copy_file3)
        self.cur_path_4 = os.path.join(macro.SCRIPTS_PATH, self.copy_file4)
        self.cur_path_5 = os.path.join(macro.SCRIPTS_PATH, self.copy_file5)
        self.parent_path = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.target_path1 = os.path.join(self.parent_path, self.copy_file1)
        self.target_path2 = os.path.join(self.parent_path, self.copy_file2)
        self.target_path3 = os.path.join(self.parent_path, self.copy_file3)
        self.target_path4 = os.path.join(self.parent_path, self.copy_file4)
        self.target_path5 = os.path.join(self.parent_path, self.copy_file5)

    def test_join(self):
        step_txt = '-----step1:创建表;expect: 创建成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""drop table if exists {self.table_name1}  CASCADE ;
                    drop table if exists {self.table_name2}  CASCADE ;
                    drop table if exists {self.table_name3}  CASCADE ;
                    drop table if exists {self.table_name4}  CASCADE ;
                    drop table if exists {self.table_name5}  CASCADE ;
            create table {self.table_name1} 
            (
                c_customer_sk             integer               not null,
                c_customer_id             char(16)              not null,
                c_current_cdemo_sk        integer                       ,
                c_current_hdemo_sk        integer                       ,
                c_current_addr_sk         integer                       ,
                c_first_shipto_date_sk    integer                       ,
                c_first_sales_date_sk     integer                       ,
                c_salutation              char(10)                      ,
                c_first_name              char(20)                      ,
                c_last_name               char(30)                      ,
                c_preferred_cust_flag     char(1)                       ,
                c_birth_day               integer                       ,
                c_birth_month             integer                       ,
                c_birth_year              integer                       ,
                c_birth_country           varchar(20)                   ,
                c_login                   char(13)                      ,
                c_email_address           char(50)                      ,
                c_last_review_date        char(10)
            )  WITH (STORAGE_TYPE=USTORE);
            create table {self.table_name2} 
            (
              cc_call_center_sk         integer               not null,
              cc_call_center_id         char(16)              not null,
              cc_rec_start_date         date                          ,
              cc_rec_end_date           date                          ,
              cc_closed_date_sk         integer                       ,
              cc_open_date_sk           integer                       ,
              cc_name                   varchar(50)                   ,
              cc_class                  varchar(50)                   ,
              cc_employees              integer                       ,
              cc_sq_ft                  integer                       ,
              cc_hours                  char(20)                      ,
              cc_manager                varchar(40)                   ,
              cc_mkt_id                 integer                       ,
              cc_mkt_class              char(50)                      ,
              cc_mkt_desc               varchar(100)                  ,
              cc_market_manager         varchar(100)                   ,
              cc_division               varchar(100)                       ,
              cc_division_name          varchar(50)                   ,
              cc_company                varchar(100)                       ,
              cc_company_name           char(50)                      ,
              cc_street_number          char(10)                      ,
              cc_street_name            varchar(60)                   ,
              cc_street_type            char(15)                      ,
              cc_suite_number           char(10)                      ,
              cc_city                   varchar(60)                   ,
              cc_county                 varchar(30)                   ,
              cc_state                  char(50)                       ,
              cc_zip                    char(20)                      ,
              cc_country                varchar(20)                   ,
              cc_gmt_offset             decimal(5,2)                  ,
              cc_tax_percentage         decimal(5,2)                  
            )  WITH (STORAGE_TYPE=USTORE);
            create table {self.table_name3} 
            (
                wp_web_page_sk            integer               not null,
                wp_web_page_id            char(16)              not null,
                wp_rec_start_date         date                          ,
                wp_rec_end_date           date                          ,
                wp_creation_date_sk       integer                       ,
                wp_access_date_sk         integer                       ,
                wp_autogen_flag           char(1)                       ,
                wp_customer_sk            integer                       ,
                wp_url                    varchar(100)                  ,
                wp_type                   char(50)                      ,
                wp_char_count             integer                       ,
                wp_link_count             integer                       ,
                wp_image_count            integer                       ,
                wp_max_ad_count           integer
             )
            WITH (STORAGE_TYPE=USTORE) partition by range(wp_rec_start_date)
            (
             partition p1 values less than('1990-01-01'),
             partition p2 values less than('1995-01-01'),
             partition p3 values less than('2000-01-01'),
             partition p4 values less than('2005-01-01'),
             partition p5 values less than('2010-01-01'),
             partition p6 values less than('2015-01-01'),
             partition p7 values less than(maxvalue)
            );
            create table {self.table_name4} 
            (
                p_promo_sk                integer               not null,
                p_promo_id                char(16)              not null,
                p_start_date_sk           integer                       ,
                p_end_date_sk             integer                       ,
                p_item_sk                 integer                       ,
                p_cost                    decimal(15,2)                 ,
                p_response_target         integer                       ,
                p_promo_name              char(50)                      ,
                p_channel_dmail           char(1)                       ,
                p_channel_email           char(1)                       ,
                p_channel_catalog         char(1)                       ,
                p_channel_tv              char(1)                       ,
                p_channel_radio           char(1)                       ,
                p_channel_press           char(1)                       ,
                p_channel_event           char(1)                       ,
                p_channel_demo            char(1)                       ,
                p_channel_details         varchar(100)                  ,
                p_purpose                 char(50)                      ,
                p_discount_active         char(100)
             ) WITH (STORAGE_TYPE=USTORE);
            create table {self.table_name5} 
            (
                cp_catalog_page_sk        integer               not null,
                cp_catalog_page_id        char(16)              not null,
                cp_start_date_sk          integer                       ,
                cp_end_date_sk            integer                       ,
                cp_department             varchar(50)                   ,
                cp_catalog_number         integer                       ,
                cp_catalog_page_number    integer                       ,
                cp_description            varchar(100)                  ,
                cp_type                   varchar(100)
             ) WITH (STORAGE_TYPE=USTORE);"""
        self.log.info(cmd_sql)
        create_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(create_result)
        self.assertTrue(
            create_result.count(self.constant.TABLE_CREATE_SUCCESS) == 5,
            '执行失败:' + step_txt)

        step_txt = '-----step2:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(step_txt)
        text = '-----step2.1: 获取当前主机的IP地址;expect: 成功-----'
        self.log.info(text)
        cmd = "ifconfig -a|grep inet6 -a2|grep broadcast|awk '{print $2}'"
        self.log.info(cmd)
        self.cur_ip = subprocess.getoutput(cmd).strip()
        self.log.info(self.cur_ip)

        text = '-----step2.2: 复制文件到数据库主机对应目录 expect: 成功-----'
        self.log.info(text)
        if self.cur_ip == self.pri_sh.node.ssh_host:
            self.log.info('当前主机是数据库主机，无需远程复制')
            cmd = f"cp {self.cur_path_1} {self.parent_path};" \
                f"cp {self.cur_path_2} {self.parent_path};" \
                f"cp {self.cur_path_3} {self.parent_path};" \
                f"cp {self.cur_path_4} {self.parent_path};" \
                f"cp {self.cur_path_5} {self.parent_path};"
            self.log.info(cmd)
            res = self.pri_sh.node.sh(cmd).result()
            self.log.info(res)
        else:
            self.com.scp_file(self.pri_root, self.copy_file1,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file2,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file3,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file4,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file5,
                              self.parent_path)
            chmod_cmd = f"chmod 777 -R {self.parent_path};" \
                        f"ls -al {self.parent_path}"
            self.log.info(chmod_cmd)
            chmod_result = self.pri_root.sh(chmod_cmd).result()
            self.log.info(chmod_result)

        text = '-----step2.3: 查看数据库主机目标路径下文件 expect: 成功-----'
        self.log.info(text)
        cmd = f"ls -l {self.parent_path}| grep -E '{self.copy_file1}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file2}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file3}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file4}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file5}';"
        self.log.info(cmd)
        res = self.pri_root.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('No such file or directory', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file1}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file2}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file3}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file4}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file5}', res, '文件拷贝失败')

        text = '-----step2.4:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(text)
        self.log.info('---------------使用copy命令插入数据------------------')
        copy_sql = f"copy {self.table_name1} from '{self.target_path1}' " \
            f"delimiters ',';" \
            f"copy {self.table_name2} from '{self.target_path2}' " \
            f"delimiters ',';" \
            f"copy {self.table_name3} from '{self.target_path3}' " \
            f"delimiters ',';" \
            f"copy {self.table_name4} from '{self.target_path4}' " \
            f"delimiters ',';" \
            f"copy {self.table_name5} from '{self.target_path5}' " \
            f"delimiters ',';"
        copy_result = self.pri_sh.execut_db_sql(copy_sql)
        self.log.info(copy_result)
        self.assertIn('COPY 60', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 300', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 11718', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 10000', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 6', copy_result, '执行失败:' + step_txt)

        step_txt = '-----step3: union与join结合 expect: 成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""
        select DISTINCT  rank() over(order by 1),
                        max(c_birth_day + (30 * c_birth_month)) over() C1,
                        min((30 * c_birth_month) - c_birth_day - 10) over() C2
          from {self.table_name1}
          join {self.table_name2}
            on c_birth_country = cc_country
         where c_birth_day > 15
         GROUP BY cube(c_birth_month, cc_country, c_birth_day)
         union 
         select p_item_sk,
               sum(p_cost) over() / count(EXTRACT(day from wp_rec_start_date)) over() as c1,
               rank() over(partition by p_start_date_sk order by 1) / count(cp_end_date_sk)
          from {self.table_name3}, {self.table_name4}, {self.table_name5}
         where p_promo_sk = wp_web_page_sk
         group by p_item_sk,
                  p_channel_dmail,
                  p_cost,
                  rollup(wp_rec_start_date,
                  p_start_date_sk)
         ORDER BY 1, 2, 3;"""
        self.log.info(cmd_sql)
        query_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(query_result)
        self.assertIn('60 rows', query_result, '执行失败:' + step_txt)

    def tearDown(self):
        step_txt = '-----step4: 环境清理 expect:清理成功-----'
        self.log.info(step_txt)
        self.log.info('删除表')
        cmd_sql = self.pri_sh.execut_db_sql(f"""
                    drop table {self.table_name1} cascade;
                    drop table {self.table_name2} cascade;
                    drop table {self.table_name3} cascade;
                    drop table {self.table_name4} cascade;
                    drop table {self.table_name5} cascade;""")
        self.log.info(cmd_sql)
        text = '-----删除文件-----'
        self.log.info(text)
        rm_cmd = f"rm -rf {self.target_path1};" \
            f"rm -rf {self.target_path2};" \
            f"rm -rf {self.target_path3};" \
            f"rm -rf {self.target_path4};" \
            f"rm -rf {self.target_path5};"
        self.log.info(rm_cmd)
        self.com.get_sh_result(self.pri_root, rm_cmd)
        for s in [self.target_path1, self.target_path2]:
            cmd = f'if [ -f {s} ]; then echo "does exists"; ' \
                f'else echo "not exists"; fi'
            file_judgments = self.com.get_sh_result(self.pri_root, cmd)
            self.assertIn('not exists', file_judgments, "清理失败" + text)
        self.assertTrue(
            cmd_sql.count(self.constant.DROP_TABLE_SUCCESS) == 5,
            '执行失败:' + step_txt)
        self.log.info(f'----- {os.path.basename(__file__)} end-----')
